package com.zhang.gmall.realtime.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.zhang.gmall.realtime.beans.TableProcess;
import com.zhang.gmall.realtime.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Arrays;
import java.util.List;

/**
 * @title: 动态分流逻辑
 * @author: zhang
 * @date: 2022/3/4 19:14
 */
public class TableProcessFunc extends BroadcastProcessFunction<JSONObject, String, JSONObject> {
    private OutputTag<JSONObject> dimTag;
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;
    private Connection connection;


    public TableProcessFunc(OutputTag<JSONObject> dimTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.dimTag = dimTag;
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //注册驱动
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        //创建连接
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    //处理广播流数据
    //{"database":"","data":{},"type":"insert","table":"table_process"}
    @Override
    public void processBroadcastElement(String jsonStr, BroadcastProcessFunction<JSONObject, String, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
        //todo 1 解析参数,封装tableProcess对象，封装状态放入广播
        //将json字符串转化为obj
        JSONObject jsonObject = JSON.parseObject(jsonStr);
        //获取data属性，配置表的一条配置记录
        JSONObject data = jsonObject.getJSONObject("data");
        //转化为java对象
        TableProcess tableProcess = JSON.toJavaObject(data, TableProcess.class);
        //根据配置对象获取相关信息
        String sourceTable = tableProcess.getSourceTable();
        String operateType = tableProcess.getOperateType();
        // 封装状态广播出去
        String key = sourceTable + "-" + operateType;
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        broadcastState.put(key, tableProcess);
        String sinkType = tableProcess.getSinkType();
        //todo 3 如果输出类型为hbase则需要创建Phoenix表
        if (TableProcess.SINK_TYPE_HBASE.equals(sinkType) && "insert".equals(operateType)) {
            //创建Phoenix表
            createTable(
                    tableProcess.getSinkTable(),
                    tableProcess.getSinkColumns(),
                    tableProcess.getSinkPk(),
                    tableProcess.getSinkExtend()
            );
        }
    }


    //处理业务数据
    @Override
    public void processElement(JSONObject jsonObj, BroadcastProcessFunction<JSONObject, String, JSONObject>.ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //todo 2 获取广播状态数据
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        //封装key
        String table = jsonObj.getString("table");
        String type = jsonObj.getString("type");


        //注意：如果使用maxwell的bootstrap同步历史数据的话，操作类型为bootstrap-insert，需要进行修改
        if (type.equals("bootstrap-insert")) {
            type = "insert";
            jsonObj.put("type", type);
        }

        String key = table + "-" + type;
        TableProcess tableProcess = broadcastState.get(key);

        //todo 4 判断状态数据是否为空
        if (tableProcess != null) {
            //说明配置表有要处理的表信息
            String sinkTable = tableProcess.getSinkTable();
            //维度数据和事实数据向下游传递之前都需要携带目的地信息 sinkTable
            jsonObj.put("sinkTable", sinkTable);
            //根据配置表过滤字段
            String sinkColumns = tableProcess.getSinkColumns();
            JSONObject data = jsonObj.getJSONObject("data");
            filterColumn(data, sinkColumns);

            String sinkType = tableProcess.getSinkType();
            if (TableProcess.SINK_TYPE_HBASE.equals(sinkType)) {
                ctx.output(dimTag, jsonObj);
            } else if (TableProcess.SINK_TYPE_KAFKA.equals(sinkType)) {
                out.collect(jsonObj);
            }
        } else {
            //说明配置表没有要处理的表信息
            System.out.println("该组合key：" + key + "不存在！");
        }

    }

    @Override
    public void close() throws Exception {
        connection.close();
    }

    /**
     * 创建Phoenix表
     *
     * @param sinkTable   表名
     * @param sinkColumns 字段
     * @param sinkPk      主键
     * @param sinkExtend  扩展字段
     */
    private void createTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        PreparedStatement statement = null;
        try {
            if (sinkPk == null) {
                sinkPk = "id";
            }
            if (sinkExtend == null) {
                sinkExtend = "";
            }else {
                sinkExtend = " , "+sinkExtend;
            }

            StringBuilder creatTableSQL = new StringBuilder("CREATE TABLE IF NOT EXISTS ")
                    .append(GmallConfig.HBASE_SCHEMA)
                    .append(".")
                    .append(sinkTable)
                    .append("(");
            String[] fields = sinkColumns.split(",");

            for (int i = 0; i < fields.length; i++) {
                String field = fields[i];
                //判断是否为主键
                if (field.equals(sinkPk)) {
                    creatTableSQL.append(field).append(" VARCHAR primary key ");
                } else {
                    creatTableSQL.append("info").append(".").append(field).append(" VARCHAR ");
                }
                if (i < fields.length - 1) {
                    creatTableSQL.append(",");
                }
            }
            creatTableSQL.append(")").append(" column_encoded_bytes=0").append(sinkExtend);
            System.out.println("Phoenix建表语句:" + creatTableSQL);

            statement = connection.prepareStatement(creatTableSQL.toString());
            statement.execute();
        } catch (SQLException e) {
            throw new RuntimeException("Phoenix建表失败！");
        } finally {
            if (statement != null) {
                try {
                    statement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 根据配置表字段过滤字段
     *
     * @param data
     * @param sinkColumns
     */
    private void filterColumn(JSONObject data, String sinkColumns) {
        String[] fields = sinkColumns.split(",");
        List<String> fieldsList = Arrays.asList(fields);
        data.keySet().removeIf(field -> !fieldsList.contains(field));
    }


}
